package co.yixiang.yshop.framework.mq.redis.core;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;

/**
 * Redis Stream 初始化工具类
 * 
 * @author yshop
 */
@Slf4j
public class RedisStreamInitializer {

    /**
     * 安全地创建 Stream 和 Consumer Group
     * 
     * @param redisTemplate Redis模板
     * @param streamKey Stream键
     * @param groupName 消费者组名
     * @return 是否成功
     */
    public static boolean safeCreateStreamAndGroup(RedisTemplate<String, ?> redisTemplate, 
                                                 String streamKey, 
                                                 String groupName) {
        try {
            // 1. 检查Stream是否存在，如果不存在则创建
            if (!redisTemplate.hasKey(streamKey)) {
                log.info("[RedisStreamInitializer][StreamKey({}) 不存在，正在创建...]", streamKey);
                // 创建一个空消息来初始化Stream
                redisTemplate.opsForStream().add(ObjectRecord.create(streamKey, ""));
                log.info("[RedisStreamInitializer][成功创建Stream({})]", streamKey);
            }

            // 2. 创建Consumer Group
            try {
                redisTemplate.opsForStream().createGroup(streamKey, groupName);
                log.info("[RedisStreamInitializer][成功创建Consumer Group({}) 对应的Stream({})]", 
                        groupName, streamKey);
                return true;
            } catch (Exception e) {
                // 如果是Consumer Group已存在的错误，可以忽略
                if (e.getMessage() != null && e.getMessage().contains("BUSYGROUP")) {
                    log.info("[RedisStreamInitializer][Consumer Group({}) 已存在，忽略创建错误]", groupName);
                    return true;
                }
                log.warn("[RedisStreamInitializer][创建Consumer Group失败，StreamKey({}), Group({}), 错误: {}]",
                        streamKey, groupName, e.getMessage());
                return false;
            }
        } catch (Exception e) {
            log.error("[RedisStreamInitializer][初始化Stream和Consumer Group失败，StreamKey({}), Group({})]", 
                    streamKey, groupName, e);
            return false;
        }
    }

    /**
     * 检查 Stream 和 Consumer Group 是否存在
     * 
     * @param redisTemplate Redis模板
     * @param streamKey Stream键
     * @param groupName 消费者组名
     * @return 是否都存在
     */
    public static boolean checkStreamAndGroupExists(RedisTemplate<String, ?> redisTemplate, 
                                                  String streamKey, 
                                                  String groupName) {
        try {
            // 检查Stream是否存在
            boolean streamExists = redisTemplate.hasKey(streamKey);
            if (!streamExists) {
                log.warn("[RedisStreamInitializer][StreamKey({}) 不存在]", streamKey);
                return false;
            }

            // 检查Consumer Group是否存在
            try {
                redisTemplate.opsForStream().groups(streamKey);
                log.info("[RedisStreamInitializer][StreamKey({}) 和 Consumer Group({}) 都存在]", 
                        streamKey, groupName);
                return true;
            } catch (Exception e) {
                log.warn("[RedisStreamInitializer][Consumer Group({}) 不存在或查询失败]", groupName);
                return false;
            }
        } catch (Exception e) {
            log.error("[RedisStreamInitializer][检查Stream和Consumer Group失败]", e);
            return false;
        }
    }
} 